// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/crypto_common/feed_v2/processor.h"

#include <string>
#include <string_view>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/crypto_common/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::crypto_common::feed_v2::impl {

CryptoFeedProcessor::CryptoChannel
ParseChannel(const rapidjson::Document& doc) {
  CryptoFeedProcessor::CryptoChannel info;
  if (doc.HasMember("url")) {
    if (doc["data"]["code"].GetString() != std::string_view("0")) {
      return info;
    }
    info.channel = doc["data"]["method"].GetString();
    info.parameters =  base::feed::FeedParsingProcessor::GetRequestParameters(
        doc["url"].GetString());
    if (info.channel == "public/get-candlestick") {
      info.native_symbol = info.parameters["instrument_name"];
      info.channel_sign = "?instrument_name={}&timeframe=" + info.parameters["timeframe"];
    } else {
      THROW() << "NOT support url" << doc["url"].GetString();
    }
  } else {
    info.channel = doc["result"]["channel"].GetString();
    info.channel_sign = info.channel;
    info.native_symbol = doc["result"]["instrument_name"].GetString();
  }
  return info;
}

void CryptoFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);
  rapidjson::Document document;
  if(packet.payload[0] == '\0'){
    return;
  }
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError())
            << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (!document.HasMember("result") && !document.HasMember("url") ) {
    LOG(INFO) << "[Crypto Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
    return;
  }

  auto info = ParseChannel(document);
  if (!CheckChannelFilter(info.channel_sign)) {
    return;
  }

  const std::string& native_symbol = info.native_symbol;
  if (StringStartsWith(info.channel, "trade")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    int index = 1;
    int last = document["result"]["data"].Size();
    for (const auto& trade_json : document["result"]["data"].GetArray()) {
      if (!(trade_json.HasMember("s") && trade_json.HasMember("p") && trade_json.HasMember("q"))) {
        continue;
      }
      if (trade_json["s"].IsNull() || trade_json["p"].IsNull() || trade_json["q"].IsNull()) {
        continue;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      trade->Clear();
      CryptoFeedUpdateParser::ParseTrade(trade_json, trade);
      upd_mgr_.PublishTrade(native_symbol, index != last);
      index++;
    }
  } else if (StringStartsWith(info.channel, "book")) {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      for(const auto& book_json : document["result"]["data"].GetArray()) {
        CryptoFeedUpdateParser::ParseBook(
          book,
          book_json);
        upd_mgr_.PublishBook();
      }
    }
  } else if (StringStartsWith(info.channel, "index")) {
    // index symbol format: BTCUSD-INDEX
    std::vector<std::string> splitted;
    splitted = Split(native_symbol, "-");
    CHECK_EQ(splitted.size(), 2);
    CHECK_EQ(splitted[1], "INDEX");
    std::string& underlying_symbol = splitted[0];
    auto* index = upd_mgr_.GetIndexMutable();

    for (const auto& index_json : document["result"]["data"].GetArray()) {
      CryptoFeedUpdateParser::ParseIndex(index_json, index);

      for (const auto& product : upd_mgr_.products()) {
        if (product->native_symbol().find(underlying_symbol) != std::string::npos) {
          upd_mgr_.PublishIndexWithExchangeTimeCheck(product->feed_native_symbol());
        }
      }
    }
  } else if (StringStartsWith(info.channel, "mark")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    auto* mark_price = upd_mgr_.GetMarkPriceMutable();
    for (const auto& mark_json : document["result"]["data"].GetArray()) {
      CryptoFeedUpdateParser::ParseMarkPrice(mark_json, mark_price);
      upd_mgr_.PublishMarkPriceWithExchangeTimeCheck(native_symbol);
    }
  } else if (StringStartsWith(info.channel, "funding")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    auto* funding_rate = upd_mgr_.GetFundingRateMutable();
    for (const auto& funding_json : document["result"]["data"].GetArray()) {
      CryptoFeedUpdateParser::ParseFundingRate(funding_json, funding_rate);
      upd_mgr_.PublishFundingRateWithExchangeTimeCheck(native_symbol);
    }
  } else if (StringStartsWith(info.channel, "ticker")) {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    auto* open_interest = upd_mgr_.GetOpenInterestMutable();
    for (const auto& open_interest_json : document["result"]["data"].GetArray()) {
      if (document.HasMember("oi") && !open_interest_json["oi"].IsNull()) {
        CryptoFeedUpdateParser::ParseOpenInterest(open_interest_json, open_interest);
        upd_mgr_.PublishOpenInterestWithExchangeTimeCheck(native_symbol);
      }
    }
  } else if (info.channel == "public/get-candlestick") {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    auto interval = CryptoFeedUpdateParser::ParseKlineInterval(
        document["data"]["result"]["interval"].GetString());
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      auto data = document["data"]["result"]["data"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = stoll(data[0]["t"].GetString()) * 1'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = CryptoFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    LOG(INFO) << "[Crypto Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::crypto_common::feed_v2::impl
